package mqtt

import (
	"context"
	"fmt"
	pmqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/rs/zerolog"
	"sync"
	"time"
)

type Server struct {
	Log     *zerolog.Logger
	ctx     context.Context
	cancel  context.CancelFunc
	mqtt    pmqtt.Client
	lock    sync.RWMutex
	clients map[ClientID]*Client
	do      chan struct{}
}

func (s *Server) subscribe(topic string, qos byte, callback pmqtt.MessageHandler) error {
	token := s.mqtt.Subscribe(topic, qos, callback)
	token.Wait()
	return token.Error()
}

func (s *Server) publish(topic string, qos byte, retained bool, payload interface{}) error {
	token := s.mqtt.Publish(topic, qos, retained, payload)
	token.Wait()
	return token.Error()
}

func (s *Server) register(subscribes []Subscribe) {
	for _, sub := range subscribes {
		e := s.subscribe(sub.Topic, sub.Qos, sub.Handler(s))
		if e != nil {
			e = fmt.Errorf("subscribe Topic:%s err:%s", sub.Topic, e)
			s.Log.Error().Msg(e.Error())
			panic(e)
		}
		s.Log.Debug().Msgf("Subscribe Topic:%s success", sub.Topic)
	}
}

func (s *Server) GetClient(id ClientID) *Client {
	s.lock.RLocker()
	//开启读锁,读取
	c, ok := s.clients[id]
	s.lock.RUnlock()

	//存在则直接返回
	if ok {
		return c
	} else {
		//不存在,则开启读写锁,然后在读取一次(防止并发读没有,重复写入),还没有就重新创建
		s.lock.Lock()
		defer s.lock.Unlock()
		if c, ok := s.clients[id]; ok {
			return c
		} else {
			c := newClient(c.ctx, c.log, id)
			s.clients[id] = c
			return c
		}
	}
}

func (s *Server) expireClient() {
	//先开启读锁,然后读取过期的客户端id并存储
	s.lock.RLock()
	t := time.Now()
	expireId := make([]ClientID, 0)
	for _, c := range s.clients {
		if t.Sub(c.time) > defaultClientTtl {
			expireId = append(expireId, c.clientId)
		}
	}
	s.lock.RUnlock()

	//然后开启读写锁,然后过期客户端
	if len(expireId) > 0 {
		s.lock.Lock()
		defer s.lock.Unlock()
		for _, id := range expireId {
			if c, ok := s.clients[id]; ok {
				//取消客户端
				c.cancel()
				delete(s.clients, id)
			}
		}
	}
	s.Log.Debug().Msg("定时清理过期的客户端")
}

func (s *Server) clearClient() {
	s.lock.Lock()
	defer s.lock.Unlock()
	s.clients = make(map[ClientID]*Client)
	s.Log.Debug().Msg("清空客户端")
}

func (s *Server) start(subscribes []Subscribe) {
	defer func() {
		s.Log.Info().Msg("mqtt close")
		close(s.do)
	}()

	//注册订阅
	s.register(subscribes)

	ticker := time.NewTicker(defaultClientCycleDuration)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C: //定时清除过期客户端
			s.expireClient()

		case <-s.ctx.Done(): //停止,清空客户端
			s.clearClient()
			return
		}
	}
}

func (s *Server) Stop() {
	s.cancel()
	<-s.do
}

func (s *Server) getReply(ctx context.Context, c *Client, msgId MsgID, replyChan chan<- *CmdReply) {
	for {
		select {
		case <-ctx.Done(): //结束,获取错误信息
			replyChan <- NewCmdReply(false, ctx.Err())
			return

		default:
			r := c.ReadReply(msgId)
			if r != nil {
				replyChan <- r
				return
			}
		}
	}
}

func (s *Server) Publish(ctx context.Context, req *RemotingRequest) (r *MsgReply, e error) {
	e = s.publish(req.Topic, req.Qos, req.Retained, req.Payload)
	if e != nil {
		return
	}

	//初始化回复信息
	r = &MsgReply{
		Msg: Msg{
			MsgID:    req.Payload.MsgID,
			ClientId: req.Payload.ClientId,
			Reply:    req.Payload.Reply,
		},
	}

	//获取客户端
	c := s.GetClient(r.ClientId)
	//更新客户端时间
	c.time = time.Now()

	//如果消息需要消息回复,等待回复,如果超时则为失败
	if req.Payload.Reply {
		replyChan := make(chan *CmdReply)
		defer close(replyChan)

		ctx, cancel := context.WithTimeout(ctx, defaultMaxReplyTime)
		defer cancel()

		//读取回复
		go s.getReply(ctx, c, r.MsgID, replyChan)

		for {
			select {
			case <-ctx.Done(): //结束,获取错误信息
				r.Data = NewCmdReply(false, ctx.Err())
				return

			case r.Data = <-replyChan: //规定时间内返回了结果
				return
			}
		}
	} else {
		//不需要等待回复,则直接成功
		r.Data = NewCmdReply(true, nil)
	}

	return
}

func NewService(conf Conf, log *zerolog.Logger, subscribes []Subscribe) (s *Server, e error) {
	c, err := newMqtt(conf)
	if err != nil {
		e = err
		return
	}
	l := log.With().Str("type", "deviceServer").Logger()
	l.Info().Msgf("mqtt: %s connect success", conf.Dns)

	cCtx, cancel := context.WithCancel(context.Background())
	s = &Server{
		Log:     &l,
		ctx:     cCtx,
		cancel:  cancel,
		mqtt:    c,
		lock:    sync.RWMutex{},
		clients: make(map[ClientID]*Client),
		do:      make(chan struct{}),
	}
	go s.start(subscribes)

	return
}
